﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team - cellcloudproject@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellExecutorListener.h"
#include "CellThreadPoolExecutor.h"
#include "CellUtil.h"
#include "CellLoggerManager.h"

namespace cell
{

#define DEFAULT_CORE_SIZE 8
#define DEFAULT_MAX_SIZE 64
#define WARMUP_NUM 1

//-----------------------------------------------------------------------
ThreadPoolExecutor::ThreadPoolExecutor(const String& name, uint32 maximumPoolSize)
	: ExecutorService(name)
{
	construct(maximumPoolSize, maximumPoolSize);
}
//-----------------------------------------------------------------------
ThreadPoolExecutor::ThreadPoolExecutor(const String& name, uint32 corePoolSize,
	uint32 maximumPoolSize)
	: ExecutorService(name)
{
	construct(corePoolSize, maximumPoolSize);
}
//-----------------------------------------------------------------------
ThreadPoolExecutor::~ThreadPoolExecutor(void)
{
	CELL_LOCK_MUTEX(mMutex);

	TaskList::iterator iter, end = mTaskList.end();
	for (iter = mTaskList.begin(); iter != end; ++iter)
	{
		CELL_DELETE (*iter);
	}
	mTaskList.clear();

	end = mExpiredTaskList.end();
	for (iter = mExpiredTaskList.begin(); iter != end; ++iter)
	{
		CELL_DELETE (*iter);
	}
	mExpiredTaskList.clear();
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::construct(uint32 corePoolSize,
	uint32 maximumPoolSize)
{
	mCorePoolSize = corePoolSize;
	mMaximumPoolSize = maximumPoolSize;

	if (mCorePoolSize == 0)
		mCorePoolSize = DEFAULT_CORE_SIZE;
	if (mMaximumPoolSize == 0)
		mMaximumPoolSize = DEFAULT_MAX_SIZE;

	if (mMaximumPoolSize < mCorePoolSize)
		mMaximumPoolSize = mCorePoolSize;

	// 启动默认自旋
	warmup();
}
//-----------------------------------------------------------------------
inline uint32 ThreadPoolExecutor::getMaxPoolSize(void)
{
	return mMaximumPoolSize;
}
//-----------------------------------------------------------------------
inline size_t ThreadPoolExecutor::getTaskListSize(void)
{
	CELL_LOCK_MUTEX(mMutex);
	return mTaskList.size();
}
//-----------------------------------------------------------------------
inline size_t ThreadPoolExecutor::getActiveCount(void)
{
	CELL_LOCK_MUTEX(mMutex);
	return mActiveTaskList.size();
}
//-----------------------------------------------------------------------
inline size_t ThreadPoolExecutor::getIdleCount(void)
{
	CELL_LOCK_MUTEX(mMutex);
	return mIdleTaskList.size();
}
//-----------------------------------------------------------------------
size_t ThreadPoolExecutor::getTaskCount(void)
{
	CELL_LOCK_MUTEX(mMutex);
	return mTaskTable.size();
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::addListener(ExecutorListener* listener)
{
	CELL_LOCK_MUTEX(mListenerMutex);

	ListenerList::iterator iter = std::find(mListenerList.begin(), mListenerList.end(), listener);
	if (iter == mListenerList.end())
	{
		mListenerList.push_back(listener);
	}
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::removeListener(ExecutorListener* listener)
{
	CELL_LOCK_MUTEX(mListenerMutex);

	ListenerList::iterator iter = std::find(mListenerList.begin(), mListenerList.end(), listener);
	if (iter != mListenerList.end())
	{
		mListenerList.erase(iter);
	}
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::remove(Runnable* task)
{
	if (NULL == task)
		return;

	CELL_LOCK_MUTEX(mMutex);

	TaskTable::iterator iter = std::find(mTaskTable.begin(), mTaskTable.end(), task);
	if (iter != mTaskTable.end())
	{
		mTaskTable.erase(iter);
	}
}
//-----------------------------------------------------------------------
bool ThreadPoolExecutor::execute(Runnable* task)
{
	if (NULL == task)
	{
		return false;
	}

	CELL_LOCK_MUTEX(mMutex);

	mTaskTable.push_back(task);

	// 启动线程
	return knock();
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::awaitExecution(long timeout, uint8 unit)
{
	uint64 threshold = TimeUnit::convertMillisecond(timeout, unit);
	uint64 startTime = Util::currentTimeMillis();

	bool over = false;

	while (false == over)
	{
		{
			CELL_LOCK_MUTEX(mMutex);

			ExecutorTask* task = NULL;
			TaskList::iterator iter, end = mTaskList.end();
			for (iter = mTaskList.begin(); iter != end; ++iter)
			{
				task = (*iter);
				if (task->isRunning())
					task->kill();
			}
		}

		Thread::sleep(5);

		{
			// 唤醒所有线程
			CELL_LOCK_MUTEX(mMutex);
			CELL_THREAD_NOTIFY_ALL(mNewTask);
		}

		{
			CELL_LOCK_MUTEX(mMutex);

			uint32 count = 0;
			ExecutorTask* task = NULL;
			TaskList::iterator iter, end = mTaskList.end();
			for (iter = mTaskList.begin(); iter != end; ++iter)
			{
				task = (*iter);
				if (task->isRunning())
					++count;
			}
			if (count == 0)
			{
				over = true;
				break;
			}
		}

		if (timeout < 0)
			continue;

		if (Util::currentTimeMillis() - startTime >= threshold)
			break;
	}
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::warmup(void)
{
	CELL_LOCK_MUTEX(mMutex);

	for (int i = 0; i < WARMUP_NUM; ++i)
	{
		ExecutorTask* et = CELL_NEW ExecutorTask(this, true);
		if (et->start())
		{
			mTaskList.push_back(et);
			mIdleTaskList.push_back(et);
		}
		else
		{
			CELL_DELETE et;
		}
	}
}

//-----------------------------------------------------------------------
bool ThreadPoolExecutor::knock(void)
{
	CELL_LOCK_MUTEX(mMutex);

	// 如果任务表里还有任务，而空闲线程数量不足，则尝试创建新线程

	bool ret = false;

	if (mIdleTaskList.size() == 0)
	{
		if (mTaskList.size() < mMaximumPoolSize)
		{
			// 填充核心池
			bool spinning = mCorePoolSize > mTaskList.size() ? true : false;
			ExecutorTask* et = CELL_NEW ExecutorTask(this, spinning);
			if (et->start())
			{
				mTaskList.push_back(et);
				mIdleTaskList.push_back(et);
				ret = true;

				char log[256] = {0x0};
				sprintf(log, "[CC] 线程池 %s 创建新线程成功 - 任务数：%d，线程总数：%d，活动线程数：%d，空闲线程数：%d",
					getName().c_str(), mTaskTable.size(), mTaskList.size(), mActiveTaskList.size(), mIdleTaskList.size());
				LoggerManager::getSingleton().log(log);
			}
			else
			{
				CELL_DELETE et;
				et = NULL;

			#if CELL_DEBUG_MODE
				char log[256] = {0x0};
				sprintf(log, "[CC] 线程池 %s 创建新线程失败 - 任务数：%d，线程总数：%d，活动线程数：%d，空闲线程数：%d",
					getName().c_str(), mTaskTable.size(), mTaskList.size(), mActiveTaskList.size(), mIdleTaskList.size());
				LoggerManager::getSingleton().log(log);
			#endif // CELL_DEBUG_MODE
			}
		}
		else
		{
			// 已经达到最大线程数
			char log[256] = {0x0};
			sprintf(log, "[CC] 线程池 %s 已经达到最大线程 %d 个线程的限制。", getName().c_str(), mMaximumPoolSize);
			LoggerManager::getSingleton().log(log);
		}
	}
	else
	{
		ret = true;

		CELL_THREAD_NOTIFY_ONE(mNewTask);
	}

	return ret;
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::discard(ExecutorTask* task)
{
	CELL_LOCK_MUTEX(mMutex);

	uint64 time = Util::currentTimeMillis();

	TaskList removeList;
	TaskList::iterator iter, end = mExpiredTaskList.end();
	for (iter = mExpiredTaskList.begin(); iter != end; ++iter)
	{
		ExecutorTask* et = (*iter);
		if (false == et->isRunning()
			&& et->hasWorked()
			&& et->timestamp > 0
			&& (time - et->timestamp) >= 2000)
		{
			removeList.push_back(et);
		}
	}

#if CELL_DEBUG_MODE
	if (!removeList.empty())
	{
		char log[128] = {0x0};
		sprintf(log, "[CC] 线程池 %s 销毁线程 - 任务数：%d，线程总数：%d，可销毁线程数：%d",
			getName().c_str(), mTaskTable.size(), mTaskList.size(), removeList.size());
		LoggerManager::getSingleton().log(log);
	}
#endif // CELL_DEBUG_MODE

	end = removeList.end();
	for (iter = removeList.begin(); iter != end; ++iter)
	{
		TaskList::iterator titer = std::find(mExpiredTaskList.begin(), mExpiredTaskList.end(), (*iter));
		if (titer != mExpiredTaskList.end())
			mExpiredTaskList.erase(titer);

		CELL_DELETE (*iter);
	}

	if (NULL == task)
	{
		return;
	}

	// 放入过期列表
	iter = std::find(mExpiredTaskList.begin(), mExpiredTaskList.end(), task);
	if (iter == mExpiredTaskList.end())
		mExpiredTaskList.push_back(task);

	// 从任务表和空闲表里删除
	ThreadPoolExecutor::TaskList::iterator find = std::find(mTaskList.begin(), mTaskList.end(), task);
	if (find != mTaskList.end())
	{
		mTaskList.erase(find);
	}
	find = std::find(mIdleTaskList.begin(), mIdleTaskList.end(), task);
	if (find != mIdleTaskList.end())
	{
		mIdleTaskList.erase(find);
	}
	find = std::find(mActiveTaskList.begin(), mActiveTaskList.end(), task);
	if (find != mActiveTaskList.end())
	{
		mActiveTaskList.erase(find);
	}
}
//-----------------------------------------------------------------------
bool ThreadPoolExecutor::resetCoreSize(uint32 size)
{
	if (size == 0)
		return false;

	mCorePoolSize = size;

	if (mMaximumPoolSize < mCorePoolSize)
		mMaximumPoolSize = mCorePoolSize;

	return true;
}
//-----------------------------------------------------------------------
bool ThreadPoolExecutor::resetMaxSize(uint32 size)
{
	if (size == 0 || size <= mCorePoolSize)
		return false;

	mMaximumPoolSize = size;

	return true;
}
//-----------------------------------------------------------------------
bool ThreadPoolExecutor::isActiveTask(Runnable* task)
{
	CELL_LOCK_MUTEX(mMutex);

	TaskList::const_iterator iter, end = mActiveTaskList.end();
	for (iter = mActiveTaskList.begin(); iter != end; ++iter)
	{
		ExecutorTask* et = (*iter);
		if (et->getWorker() == task)
		{
			return true;
		}
	}

	return false;
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::fireBefore(ExecutorTask* task)
{
	{
		CELL_LOCK_MUTEX(mMutex);

		TaskList::iterator iter = std::find(mIdleTaskList.begin(), mIdleTaskList.end(), task);
		if (iter != mIdleTaskList.end())
		{
			mIdleTaskList.erase(iter);
		}

		mActiveTaskList.push_back(task);
	}

	beforeExecute(task->getWorker());

	{
		CELL_LOCK_MUTEX(mListenerMutex);

		ListenerList::iterator iter, end = mListenerList.end();
		for (iter = mListenerList.begin(); iter != end; ++iter)
		{
			(*iter)->prepared(task->getWorker());
		}
	}
}
//-----------------------------------------------------------------------
void ThreadPoolExecutor::fireAfter(ExecutorTask* task)
{
	afterExecute(task->getWorker());

	{
		CELL_LOCK_MUTEX(mMutex);

		TaskList::iterator iter = std::find(mActiveTaskList.begin(), mActiveTaskList.end(), task);
		if (iter != mActiveTaskList.end())
		{
			mActiveTaskList.erase(iter);
		}

		mIdleTaskList.push_back(task);
	}

	{
		CELL_LOCK_MUTEX(mListenerMutex);

		ListenerList::iterator iter, end = mListenerList.end();
		for (iter = mListenerList.begin(); iter != end; ++iter)
		{
			(*iter)->executed(task->getWorker());
		}
	}
}
//-----------------------------------------------------------------------
Runnable* ThreadPoolExecutor::apply(void)
{
	CELL_LOCK_MUTEX(mMutex);

	if (mTaskTable.empty())
		return NULL;

	TaskTable::iterator iter = mTaskTable.begin();
	if (iter == mTaskTable.end())
		return NULL;

	Runnable* worker = (*iter);
	mTaskTable.erase(iter);
	return worker;
}


//-----------------------------------------------------------------------
ExecutorTask::ExecutorTask(ThreadPoolExecutor* executor, bool spinning)
	: mExecutor(executor)
	, mSpinning(spinning)
	, mRunning(false)
	, mWorked(false)
	, mKill(false)
	, mWorker(NULL)
	, timestamp(0)
{
}
//-----------------------------------------------------------------------
ExecutorTask::~ExecutorTask(void)
{
	Thread::~Thread();
}
//-----------------------------------------------------------------------
void ExecutorTask::run(void)
{
	mRunning = true;

	do {
		if (false == mKill)
		{
			mWorker = mExecutor->apply();
		}

		if (NULL != mWorker)
		{
			mExecutor->fireBefore(this);

			mWorker->run();

			mExecutor->fireAfter(this);

			mWorker = NULL;
		}
		else
		{
			if (false == mSpinning)
			{
				// 不自旋，退出循环
				break;
			}

			CELL_LOCK_MUTEX_NAMED(mTaskMutex, lock);
			//CELL_LOCK_MUTEX_NAMED(mExecutor->mMutex, lock);
			CELL_THREAD_WAIT(mExecutor->mNewTask, mutex, lock);
		}

		mExecutor->discard(NULL);

	} while (false == mKill);

	mWorked = true;

	mExecutor->discard(this);

	mKill = true;

	mRunning = false;

	timestamp = Util::currentTimeMillis();
}
//-----------------------------------------------------------------------
void ExecutorTask::kill(void)
{
	CELL_LOCK_MUTEX(mTaskMutex);
	//CELL_LOCK_MUTEX(mExecutor->mMutex);

	mKill = true;
	mSpinning = false;

	CELL_THREAD_NOTIFY_ONE(mExecutor->mNewTask);
}

} // end namespace cell
